package com.tom.architect.party3.day31;

class ObserverOnObservable<T> extends Observable<T> {
    Observable<T> source;
    Schedulers schedulers;
    public ObserverOnObservable(Observable<T> source, Schedulers schedulers) {
        this.source = source;
        this.schedulers = schedulers;
    }

    @Override
    protected void subscribeActual(Observer<T> observer) {
        // subscribe 需要重新包装一层
        source.subscribe(new ObserverOnObserver(observer, schedulers));
    }

    private class ObserverOnObserver<T> implements Observer<T>, Runnable {
        final Observer<T> observer; // 静态代理
        final Schedulers schedulers;
        private T item; // 缓存要传递的值

        public ObserverOnObserver(Observer<T> observer, Schedulers schedulers) {
            this.observer = observer;
            this.schedulers = schedulers;
        }

        @Override
        public void onSubscribe() {
            observer.onSubscribe();
        }

        @Override
        public void onNext(T item) throws Exception {
            this.item = item;
            schedulers.scheduleDirect(this); // 调用run。
        }

        @Override
        public void onError(Throwable e) {
            observer.onError(e);
        }

        @Override
        public void onComplete() {
            observer.onComplete();
        }

        @Override
        public void run() {
            // UI主线程 或其他策略的线程。
            try {
                observer.onNext(item);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

}
